/*
 * File:   Mailer.cpp
 * Author: henrya
 *
 * Created on June 25, 2009, 12:38 AM
 */

#include "Mailer.h"
#include "Network.h"
#include "Shell.h"
#include "ThreadCreationException.h"
#include "OSScheduler.h"
#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <iostream>
#include <sstream>
using std::stringstream;
using std::endl;

extern "C" {

    void* mailer_wrapper_func(void *args) {
        Mailer *curMailer = (Mailer *) args;
        curMailer->run();
        pthread_exit(NULL);
    }
}

Mailer::Mailer(Shell* shell,int size) : inbox(size,false), network(NULL), interupted(false),shell(shell),scheduler(NULL), mode(NON_DVR), runned(false) {
    if (pthread_mutex_init(&classLock, NULL) != 0) {
        perror("Mailer lock initialization");
        exit(1);
    }
    if (pthread_cond_init(&classCond, NULL) != 0) {
        perror("Mailer cond var initialization");
        exit(1);
    }
}

Mailer::~Mailer() {
    pthread_mutex_destroy(&classLock);
    pthread_cond_destroy(&classCond);
}

void Mailer::printRTs() {
    network->printRTs();
}

void Mailer::setNetwork(Network* network) {
    this->network = network;
}

bool Mailer::tryAddMessage(Message* message) {
    bool res=false;
    res=inbox.push(message);
    return res;
}

void Mailer::run() {
    goToSleep();
    while (!interupted) {
        switch (mode) {
        case DVR: {
            bool delivered=deliverMsgs();
            scheduler->doScheduling(delivered);
            usleep(100);
            if (dvrEnded()) {
                stringstream res;
                res<<"All nodes stable. Algorithm Summary:"<<endl;
                res<<"Total context switches: "<<scheduler->getTotalContextSwitches()<<endl;
                res<<"Total message quanta given: "<<scheduler->getTotalMsgQuantaGiven()<<endl;
                res<<"Actual message pass count: "<<inbox.getMessagePassCount()<<endl;
                res<<"Thread Table:"<<endl;
                res<<*scheduler->getThreadTableString()<<endl;
                shell->print(new string(res.str()));
                runned=true;
                mode=NON_DVR;
                scheduler->reset();
                shell->wakeUp();
                goToSleep();
            }
        }
        break;
        case NON_DVR: {
            bool delivered=deliverMsgs();
            scheduler->doScheduling(delivered);
            usleep(100);
            if (dvrEnded()) {
                scheduler->reset();
                shell->wakeUp();
                goToSleep();
            }
        }
        break;
        }
    }
    pthread_exit(NULL);
}

bool Mailer::dvrEnded() {
    bool res = (0 == inbox.size());
    res = res && network->checkAllNodesStable();
    res = res && scheduler->currThreadStable();
    return res;
}

void Mailer::start() {
    if (pthread_create(&self, NULL, &mailer_wrapper_func, (void*) this) != 0) {
        ThreadCreationException ex("Failed to create Mailer thread");
        throw ex;
    }
}

bool Mailer::deliverMsgs() {
    bool res=false;
    while (inbox.size() > 0) {
        Message* msg = inbox.pop();
        //TODO add message check system etc....
        Node* destNode = network->getNode(msg->getTo());
        destNode->addMessage(msg);
        if (mode==NON_DVR) {
            destNode->setQuanta(1+network->getNodesCount());
        }
        res=true;
    }
    return res;
}

void Mailer::setScheduler(OSScheduler* scheduler) {
    this->scheduler=scheduler;
}

void Mailer::goToSleep() {
    pthread_mutex_lock(&classLock);
    pthread_cond_wait(&classCond,&classLock);
    pthread_mutex_unlock(&classLock);
}

void Mailer::setMode(MailerMode mode){
    this->mode=mode;
}

void Mailer::interrupt(){
    interupted=true;
    wakeUp();
    join();
}

void Mailer::join() {
    if (pthread_join(self, NULL) != 0) {
        RuntimeException ex("Unable to join thread " + self);
        throw ex;
    }
}

void Mailer::wakeUp(){
    pthread_mutex_lock(&classLock);
    pthread_cond_signal(&classCond);
    pthread_mutex_unlock(&classLock);
}
bool Mailer::hasRunned(){
    return runned;
}
